package com.fourth.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.fourth.bean.TrafficPageViewBean;
import com.fourth.utils.DateFormatUtil;
import com.fourth.utils.MyClickHouseUtil;
import com.fourth.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * @author wjy
 * @create 2022-08-20 9:48
 */
public class DwsTrafficVcChArIsNewPageViewWindow {
    public static void main(String[] args) throws Exception {
        //1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //2.读取dwd层三个主题的日志数据创建流
        String pageTopic = "dwd_traffic_page_log";
        String uvTopic = "dwd_traffic_unique_visitor_detail";
        String ujTopic = "dwd_traffic_user_jump_detail";
        String groupId = "page_view_window_23";
        DataStreamSource<String> pageLogDS = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(pageTopic, groupId));
        DataStreamSource<String> uvLogDS = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(uvTopic, groupId));
        DataStreamSource<String> ujLogDS = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(ujTopic, groupId));

        //3.将三条流转成javaBean
        //处理uvLogDS
        SingleOutputStreamOperator<TrafficPageViewBean> pageViewObjWithUvDS = uvLogDS.map(line -> {
            JSONObject jsonObject = JSON.parseObject(line);
            JSONObject common = jsonObject.getJSONObject("common");
            return new TrafficPageViewBean("", "",
                    common.getString("vc"),
                    common.getString("ch"),
                    common.getString("ar"),
                    common.getString("is_new"),
                    1L, 0L, 0L, 0L, 0L, jsonObject.getLong("ts"));
        });
        //处理ujLogDS
        SingleOutputStreamOperator<TrafficPageViewBean> pageViewObjWithUjDS = ujLogDS.map(line -> {
            JSONObject jsonObject = JSON.parseObject(line);
            JSONObject common = jsonObject.getJSONObject("common");


            return new TrafficPageViewBean("", "",
                    common.getString("vc"),
                    common.getString("ch"),
                    common.getString("ar"),
                    common.getString("is_new"),
                    0L,
                    0L,
                    0L,
                    0L,
                    1L, jsonObject.getLong("ts"));
        });
        //处理pageLogDS
        SingleOutputStreamOperator<TrafficPageViewBean> pageViewObjWithPvDS = pageLogDS.map(line -> {
            JSONObject jsonObject = JSON.parseObject(line);
            JSONObject common = jsonObject.getJSONObject("common");
            JSONObject page = jsonObject.getJSONObject("page");
            String last_page_id = page.getString("last_page_id");
            return new TrafficPageViewBean("", "",
                    common.getString("vc"),
                    common.getString("ch"),
                    common.getString("ar"),
                    common.getString("is_new"),
                    0L,
                    last_page_id == null ? 1L : 0L,
                    1L,
                    page.getLong("during_time"),
                    0L, jsonObject.getLong("ts"));
        });

        //4.合并三条流 union
        DataStream<TrafficPageViewBean> unionDS = pageViewObjWithPvDS.union(pageViewObjWithUvDS, pageViewObjWithUjDS);

        //5.从合并流中提取事件时间生成WaterMark UJ数据本就开了10s的迟到窗口 + 2s 乱序 ，也就是延迟了12s 所以获取不到数据是因为，数据到的时候，窗口已经关闭了 所以这里把乱序时间改成14s就正常关联上uj数据了
        SingleOutputStreamOperator<TrafficPageViewBean> pageViewWithWmDS = unionDS.assignTimestampsAndWatermarks(WatermarkStrategy.<TrafficPageViewBean>forBoundedOutOfOrderness(Duration.ofSeconds(14)).withTimestampAssigner(new SerializableTimestampAssigner<TrafficPageViewBean>() {
            @Override
            public long extractTimestamp(TrafficPageViewBean element, long recordTimestamp) {
                return element.getTs();
            }
        }));

        //6.分组、开窗、聚合
        //将维度字段作为key分组
        KeyedStream<TrafficPageViewBean, Tuple4<String, String, String, String>> keyedStream = pageViewWithWmDS.keyBy(new KeySelector<TrafficPageViewBean, Tuple4<String, String, String, String>>() {
            @Override
            public Tuple4<String, String, String, String> getKey(TrafficPageViewBean value) throws Exception {
                return Tuple4.of(value.getVc(), value.getCh(), value.getAr(), value.getIsNew());
            }
        });
        //开窗
        WindowedStream<TrafficPageViewBean, Tuple4<String, String, String, String>, TimeWindow> window = keyedStream.window(TumblingEventTimeWindows.of(Time.seconds(10)));

        //聚合
        //   --> 增量聚合: 来一条聚合一条 ---> 计算的快,节省空间
        //   --> 全量聚合: 攒够一批再处理 ---> 适合求比率相关的指标、排序、以及获取窗口信息 缺点：延迟比较高 与实时场景有点不太符合，一般不使用

        SingleOutputStreamOperator<TrafficPageViewBean> reduceDS = window
                .reduce(new ReduceFunction<TrafficPageViewBean>() {
                            @Override
                            public TrafficPageViewBean reduce(TrafficPageViewBean value1, TrafficPageViewBean value2) throws Exception {
                                value1.setUvCt(value1.getUvCt() + value2.getUvCt());
                                value1.setSvCt(value1.getSvCt() + value2.getSvCt());
                                value1.setPvCt(value1.getPvCt() + value2.getPvCt());
                                value1.setUjCt(value1.getUjCt() + value2.getUjCt());
                                value1.setDurSum(value1.getDurSum() + value2.getDurSum());
                                return value1;
                            }
                        },
                        new WindowFunction<TrafficPageViewBean, TrafficPageViewBean, Tuple4<String, String, String, String>, TimeWindow>() {
                            @Override
                            public void apply(Tuple4<String, String, String, String> key, TimeWindow window, Iterable<TrafficPageViewBean> input, Collector<TrafficPageViewBean> out) throws Exception {
                                //通过上述过程加工，此时迭代器中形成一条数据 首先获取数据
                                TrafficPageViewBean next = input.iterator().next();

                                //补全字段
                                next.setTs(System.currentTimeMillis());
                                next.setStt(DateFormatUtil.toYmdHms(window.getStart()));
                                next.setEdt(DateFormatUtil.toYmdHms(window.getEnd()));

                                //输出数据
                                out.collect(next);
                            }
                        });

        //打印测试
        reduceDS.print("reduceDS>>>>>>>>>");

        //7.将数据写出到ClickHouse
        reduceDS.addSink(MyClickHouseUtil.getSink("insert into dws_traffic_vc_ch_ar_is_new_page_view_window values(?,?,?,?,?,?,?,?,?,?,?,?)"));

        //8.启动任务
        env.execute("DwsTrafficVcChArIsNewPageViewWindow");

    }
}
